package com.ihr360.job.core.job;

import com.ihr360.job.core.Job;
import com.ihr360.job.core.JobParameters;
import com.ihr360.job.core.UnexpectedJobExecutionException;
import com.ihr360.job.core.entity.JobExecution;
import com.ihr360.job.core.entity.StepExecution;
import com.ihr360.job.core.item.ExecutionContext;
import com.ihr360.job.core.lanuch.JobLauncher;
import com.ihr360.job.core.step.AbstractStep;
import org.springframework.util.Assert;

/**
 * A {@link Step} that delegates to a {@link Job} to do its work. This is a
 * great tool for managing dependencies between jobs, and also to modularise
 * complex step logic into something that is testable in isolation. The job is
 * executed with parameters that can be extracted from the step execution, hence
 * this step can also be usefully used as the worker in a parallel or
 * partitioned execution.
 *
 * @author Dave Syer
 *
 */
public class JobStep extends AbstractStep {

    /**
     * The key for the job parameters in the step execution context. Needed for
     * restarts.
     */
    private static final String JOB_PARAMETERS_KEY = JobStep.class.getName() + ".JOB_PARAMETERS";

    private Job job;

    private JobLauncher jobLauncher;

    private JobParametersExtractor jobParametersExtractor = new DefaultJobParametersExtractor();

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();
        Assert.state(jobLauncher != null, "A JobLauncher must be provided");
        Assert.state(job != null, "A Job must be provided");
    }

    /**
     * The {@link Job} to delegate to in this step.
     *
     * @param job a {@link Job}
     */
    public void setJob(Job job) {
        this.job = job;
    }

    /**
     * A {@link JobLauncher} is required to be able to run the enclosed
     * {@link Job}.
     *
     * @param jobLauncher the {@link JobLauncher} to set
     */
    public void setJobLauncher(JobLauncher jobLauncher) {
        this.jobLauncher = jobLauncher;
    }

    /**
     * The {@link JobParametersExtractor} is used to extract
     * {@link JobParametersExtractor} from the {@link StepExecution} to run the
     * {@link Job}. By default an instance will be provided that simply copies
     * the {@link JobParameters} from the parent job.
     *
     * @param jobParametersExtractor the {@link JobParametersExtractor} to set
     */
    public void setJobParametersExtractor(JobParametersExtractor jobParametersExtractor) {
        this.jobParametersExtractor = jobParametersExtractor;
    }

    /**
     * Execute the job provided by delegating to the {@link JobLauncher} to
     * prevent duplicate executions. The job parameters will be generated by the
     * {@link JobParametersExtractor} provided (if any), otherwise empty. On a
     * restart, the job parameters will be the same as the last (failed)
     * execution.
     *
     * @see AbstractStep#doExecute(StepExecution)
     */
    @Override
    protected void doExecute(StepExecution stepExecution) throws Exception {

        ExecutionContext executionContext = stepExecution.getExecutionContext();

        executionContext.put(STEP_TYPE_KEY, this.getClass().getName());

        JobParameters jobParameters;
        if (executionContext.containsKey(JOB_PARAMETERS_KEY)) {
            jobParameters = (JobParameters) executionContext.get(JOB_PARAMETERS_KEY);
        }
        else {
            jobParameters = jobParametersExtractor.getJobParameters(job, stepExecution);
            executionContext.put(JOB_PARAMETERS_KEY, jobParameters);
        }

        JobExecution jobExecution = jobLauncher.run(job, jobParameters);
        if (jobExecution.getStatus().isUnsuccessful()) {
            // AbstractStep will take care of the step execution status
            throw new UnexpectedJobExecutionException("Step failure: the delegate Job failed in JobStep.");
        }

    }

}
